package net.bwie.realtime.jtp.dws.log.job;

import javafx.beans.property.ReadOnlyBooleanWrapper;
import javafx.scene.control.TableFocusModel;
import jdk.nashorn.internal.ir.CallNode;
import net.bwie.realtime.jtp.dws.log.utils.AnalyzerUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import java.util.List;

/**
 * 搜索关键词实时统计，其中使用IKAnalyzer分词器进行分词,采用FlinkSQL方式数据处理
 */
public class JtpTrafficSearchKeywordMinuteWindowDwsJob {

    public static void main(String[] args) {

        //1-表执行环境
        TableEnvironment tabEnv = getTableEnv();

        //2-输入表-映射到kafka消息队列
        createInputTable(tabEnv);

        //3-数据处理-select
        Table reportTable = handle(tabEnv);

        //4-输出表-映射到doris表
        createOutputTable(tabEnv);

        //5-保存数据-insert
        saveToSink(tabEnv, reportTable);

    }

    private static TableEnvironment getTableEnv() {
        //1环境属性配置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);

        //2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");

        //3返回对象
        return tabEnv;
    }

    /**
     * FlinkSQL中自定义UDTF函数，使用IKAnalyzer分词器对搜索词进行分词
     https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/udfs/#table-functions
     */



    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE dwd_traffic_page_log_kafka_source\n" +
            "(\n" +
            "    `common` MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
            "    `page`   MAP<STRING, STRING> COMMENT '页面信息',\n" +
            "    `ts`     BIGINT,\n" +
            "    row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss.SSS')),\n" +
            "    WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE\n" +
            ") WITH (\n" +
            "    'connector' = 'kafka',\n" +
            "    'topic' = 'dwd-traffic-page-log',\n" +
            "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
            "    'properties.group.id' = 'gid_dws_traffic_search_keyword',\n" +
            "    'scan.startup.mode' = 'earliest-offset',\n" +
            "    'format' = 'json',\n" +
            "    'json.fail-on-missing-field' = 'false',\n" +
            "    'json.ignore-parse-errors' = 'true'\n" +
            ")"
        );
    }

    private static Table handle(TableEnvironment tabEnv) {
        //S1-获取搜索词和搜索时间
        Table searchLogTable = tabEnv.sqlQuery(
            "SELECT\n" +
                    "    page['item'] AS full_word\n" +
                    "    ,row_time\n" +
                    "FROM dwd_traffic_page_log_kafka_source\n" +
                    "WHERE page['item'] IS NOT NULL\n" +
                    "AND page['item_type'] = 'keyword'\n" +
                    "AND page['last_page_id'] = 'search'"
        );
        //searchLogTable.execute().print("searchLogTable");
        tabEnv.createTemporaryView("search_log_table",searchLogTable);

        //S2-使用自定义UDTF函数，对搜索词进行中文分词
        tabEnv.createTemporarySystemFunction("ik_analyzer_udtf", IKAnalyzerFunction.class);
        Table wordLogTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    full_word\n" +
                        "    ,keyword\n" +
                        "    ,row_time\n" +
                        "FROM search_log_table\n" +
                        "LATERAL TABLE(ik_analyzer_udtf(full_word)) AS T(keyword)"
        );
        //wordLogTable.execute().print("wordLogTable");
        tabEnv.createTemporaryView("word_log_table",wordLogTable);

        //S3-设置窗口进行分组，聚合计算
        Table reportTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time\n" +
                        "    ,TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time\n" +
                        "    ,keyword\n" +
                        "    ,count(keyword) AS keyword_count\n" +
                        "    ,UNIX_TIMESTAMP()*1000 AS ts\n" +
                        "FROM word_log_table\n" +
                        "GROUP BY\n" +
                        "    TUMBLE(row_time, INTERVAL '1' MINUTES),\n" +
                        "    keyword"
        );
        //reportTable.execute().print("reportTable");

        //返回计算结果
        return reportTable;
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE dws_traffic_search_keyword_window_report_doris_sink (\n" +
                        "     `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "     `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
                        "     `keyword` STRING COMMENT '搜索关键词',\n" +
                        "     `keyword_count` BIGINT COMMENT '搜索关键词被搜索次数',\n" +
                        "     `ts` BIGINT COMMENT '数据产生时间戳'\n" +
                        ") WITH (\n" +
                        "      'connector' = 'clickhouse',\n" +
                        "      'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "      'table' = 'dws_traffic_search_keyword_window_report',\n" +
                        "      'username' = 'default',\n" +
                        "      'password' = '',\n" +
                        "      'format' = 'json'\n" +
                        ")"
        );

    }

    private static void saveToSink(TableEnvironment tabEnv, Table reportTable) {
        //a-注册Table为表，给表名称
        tabEnv.createTemporaryView("report_table",reportTable);
        //b-查询输入
        tabEnv.executeSql(
                "INSERT INTO dws_traffic_search_keyword_window_report_doris_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "    ,DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "    ,keyword\n" +
                        "    ,keyword_count\n" +
                        "    ,ts\n" +
                        "FROM report_table"
        );

    }
//好了 嗯呢宝贝  晚上吃什么  好爱你你可以亲亲我的脚吗  不可以  why??? buheshi  有什么不合适 你爱我就要爱我的全部 就比如说我爱你我就爱你的毛毛 那你爱我的皮燕子吗  爱  你可以亲亲他吗 可以
//吃一下呢  吃不到  那在班里让我抱着亲你的脚 我怎么能满足你 怎么不能呢 我是不是你的女朋友 是 那有什么不方便的吗 不行 好好好你就是不爱我了 给我扣帽子 扣了一点点
//这本来就是愿意就是愿意不愿意就是不愿意 为什么能扯都爱不爱你上呢 不愿意OK知道了 晚上吃什么 土豆丝吗  想吃馒头吃馒头得抄一份 那就吃炒菜 抄一个土豆丝吗  换 西红柿鸡蛋也不错
//不喜欢吃西红柿就吃木须肉 不吃 你打开菜单看看 什么菜 有什么菜 我没记住呀  五花肉炖白菜豆腐粉条你想吃这个呀  不是  看着名字感觉好吃 尝尝吧 里面想加什么菜 这个好像可以定制 不不不 有多少呀
//能吃了吗 吃不了就浪费了 吃饭为的是吃饱 吃好 又不是为了吃完 好 吃 有没有不想吃的菜 或者想吃的菜 这个清淡一点 嗯 还选选吗  还是就这个吧 就这个吧就这个吧  馒头？ 我要吃一个半
//你晚上要吃吗 或者别的什么时候 怎么了 吃一个吧  4个 我想吃粉条你为什么要把粉条换成茄子呢 我看成你不想吃粉条了 奥奥  刚看到 咋还括弧了 谁括弧了 嗯 啥时候括的  早都扩上了
//咋了说话呀 突然不知道唠什么 了那就终止吧 想干啥干点啥吧  还不想 那你想干啥 想说说甜蜜话  多甜蜜 我要吃吃晚上的饭看看好不好吃  我吃着不错  我吃吃不会是一锅吧 你吃过炖菜吗 我不知道长什么样
//妈妈做的炖菜特好吃 我好像吃过 烩菜呢 不知道长啥样  就是有点黏糊糊的 晚上尝尝  好  又一些妈妈的味道 他家的  伺候好我  怎么伺候
//这周你还给我搓吗 你想要吗  想要呀 可以  对了 这两天有个任务 什么任务 买两身衣服 什么？ 这不是快毕业了吗  你是想买衣服是吗 我给你选吗 我是说咱俩买几件一样的衣服 现在吗
//这几天吧 这么赶吗宝贝 也不着急 嗯嗯 可以 你还记得我跟你说我有一个冲锋衣 记得  明天早上我拿着你穿一下我看看合适吗 肯定会大 买小一码 妈妈给我买的L 我穿着紧，我又让她给我换的xl
//明天试试可以吗 周六啥的有时间的吧 还得带来带去的 哎哟有啥麻烦的呀 就能早点买 可好穿了 我让妈妈给买 欸别 咋了 害怕 我可没订单 妈妈给买的 再说了妈妈又不是不知道你狗东西 怕什么呀
//不行不行  不能拒绝啊  我特害怕  墨镜咱俩买一样的吗  再议 先说衣服 怎么了衣服怎么了 我害怕 害怕什么 她又没在身边 不行不行 不行不能拒绝 听见了嘛 不许犟嘴 明天先试试吧 我可挑
//可好穿 你看了就喜欢知道吗 你一定特喜欢 相信我的眼光好吗宝贝 好 这是你的心里话是吧 ok太棒了 想着逗你一下感觉说出来不合适 挨扇了吧我看你是  你想干啥 晚上让我抓一下打着伞没人拍我们我抓一下你的勾子
//不犯毛病吧  不可 万万不可 不行不能拒绝我 你就是不想亲我不想吻我不想让我抓你的勾子 你怎么又给自己加戏了呀 是不是吧 就是你的内心独白我都度出来了都
//暂停一下 想吃水果吗晚上  你要点谁家的 还有谁家的 我就知道红红  红红不好吃 我也不知道这是谁拉我进去的 点过吗 没点过  其实水果也差不多那你点吧 吃什么 随便 想一个 不想吃 菠萝 不
//吃木瓜 你说的这个你吃吗 我不吃 你得吃 多多的吃 为啥 你不吃你让我吃 长mm的  谁说的 你看  好吧 我就怕有坏的 现在大枣里面有虫子 这个可以放几天 往哪放 桌洞里呀 不像西瓜啥的过夜就不行了
//买了你可得给我吃啊 你不吃吗  你要是就吃一两个我就得扇你 起码得吃5个 这句话温柔点说  买了要是不吃你看我给不给你的大香蕉给敲碎  ？？ 这是温柔的样子吗 是 温柔的话
//迅速麻溜点 下课 你好好的温柔的说  大枣有多少啊 别买多听见了吗  吃多少买多少  一斤   三五十个 嗯嗯 可以点吧  你还没说呢  没说什么 你好好的温柔的说  买了要是不吃你看我给不给你的小66给敲碎
//主任今天给拿烟吗 不知道啊 这个是谁呀 对面超市 那个我删的那个男的  哪个啊 欠他钱的那个  嗷嗷 不是好友没关系  say 要学习吗 你不是不说话吗 那学会儿吧 嗯


    /** 
     * FlinkSQL中自定义UDTF函数,使用IKAnalyzer分词器对搜索词进行分词
     */
    @FunctionHint(output = @DataTypeHint("ROW<keyword STRING>"))
    private static class IKAnalyzerFunction extends TableFunction<Row> {
        public void eval(String fullWord) throws Exception {
            List<String> list = AnalyzerUtil.ikAnalyzer(fullWord);
            for (String keyword : list) {
                collect(Row.of(keyword));
            }
        }
    }
}
